package org.hbhk.hms.mq.tx.provider;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.rocketmq.client.producer.SendResult;
import org.hbhk.hms.mq.tx.mq.JsonUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.core.Ordered;
import org.springframework.transaction.support.TransactionSynchronizationAdapter;
import org.springframework.transaction.support.TransactionSynchronizationManager;

public class MqTransactionManager extends TransactionSynchronizationAdapter
		implements InitializingBean, ApplicationContextAware, Ordered {

	public static ApplicationContext context;

	private static final Logger LOGGER = LoggerFactory.getLogger(MqTransactionManager.class);
	private static final ThreadLocal<List<Runnable>> RUNNABLES = new ThreadLocal<List<Runnable>>();

	private static final ThreadLocal<List<MqRequest>> mqRecordDatas = new ThreadLocal<List<MqRequest>>();

	private ThreadPoolExecutor executor;
	private int bufferSize = 16384;

	private int minConsumeThread = 20;

	private int maxConsumeThread = 40;

	@Autowired
	private IMqRecordService mqRecordService;

	@Value("${hms.mq.namesrvAddr}")
	private String namesrvAddr;

	public void execute(final MqRequest mqRequest) {

		LOGGER.debug("收集mq消息待发送...");
		Runnable runnable = new Runnable() {
			@Override
			public void run() {
				try {
					LOGGER.debug("send mq  record start ");
					SendResult sendResult = mqRecordService.sendMessage(mqRequest);
					LOGGER.debug("update  sucess mq  record start ");
					updateMqrecord(mqRequest, sendResult.toString(), 1);
					LOGGER.debug("update  sucess mq  record end ");
					LOGGER.debug("send mq  record end ");
				} catch (Exception e) {
					// 发送失败记录 mq 事务表
					LOGGER.error("mq  error info ", e);
					LOGGER.error("update fialed mq record start ");

					String err = ExceptionUtils.getMessage(e);
					if (err.length() > 500) {
						err = err.substring(0, 499);
					}
					updateMqrecord(mqRequest, err, 0);
					LOGGER.error("update fialed mq record end ");

				}
			}
		};
		LOGGER.debug("submitting new runnable {} to run after commit", runnable);
		List<Runnable> threadRunnables = RUNNABLES.get();
		if (threadRunnables == null) {
			threadRunnables = new ArrayList<Runnable>();
			RUNNABLES.set(threadRunnables);
			TransactionSynchronizationManager.registerSynchronization(this);
		}
		if (mqRecordDatas.get() == null) {
			List<MqRequest> mqRequests = new ArrayList<MqRequest>();
			mqRecordDatas.set(mqRequests);
		}
		mqRecordDatas.get().add(mqRequest);

		threadRunnables.add(runnable);
	}

	@Override
	public void beforeCommit(boolean readOnly) {
		txInsert();
	}

	public void txInsert() {
		List<MqRequest> mqRequests = mqRecordDatas.get();

		List<MqRecord> mqRecords = new ArrayList<MqRecord>();
		for (MqRequest mqRequest : mqRequests) {
			MqRecord mqRecord = new MqRecord();
			mqRecord.setId(mqRequest.getReqId());
			mqRecord.setErrMsg(null);
			mqRecord.setRequest(JsonUtil.objectToJson(mqRequest));
			mqRecord.setShard(mqRequest.getMsgKey());
			mqRecord.setTopic(mqRequest.getTopic());
			mqRecord.setStatus(0);
			mqRecord.setTable(mqRequest.getTxTable());
			mqRecord.setCreateTime(new Date());
			mqRecord.setRetries(0);
			mqRecords.add(mqRecord);
		}
		mqRecordService.insertBatch(mqRecords);
	}

	public void updateMqrecord(MqRequest mqRequest, String errMsg, Integer status) {
		MqRecord mqRecord = new MqRecord();
		mqRecord.setId(mqRequest.getReqId());
		mqRecord.setTable(mqRequest.getTxTable());
		mqRecord.setShard(mqRequest.getMsgKey());
		mqRecord.setStatus(status);
		mqRecord.setErrMsg(errMsg);
		mqRecordService.update(mqRecord);

	}

	@Override
	public void afterCommit() {
		List<Runnable> runnables = RUNNABLES.get();
		for (Runnable runnable : runnables) {
			LOGGER.debug("Executing runnable {}", runnable);
			try {
				executor.submit(runnable);
			} catch (RuntimeException e) {
				LOGGER.error("failed to execute runnable " + runnable, e);
			}
		}
	}

	@Override
	public void afterCompletion(int status) {
		LOGGER.debug("transaction completed with status {}", status == STATUS_COMMITTED ? "COMMITTED" : "ROLLED_BACK");
		RUNNABLES.set(null);
		mqRecordDatas.set(null);
	}

	@Override
	public void afterPropertiesSet() throws Exception {
		BlockingQueue<Runnable> blockingQueue = new ArrayBlockingQueue<Runnable>(bufferSize);
		executor = new ThreadPoolExecutor(minConsumeThread, maxConsumeThread, 60000L, TimeUnit.MILLISECONDS,
				blockingQueue);
	}

	@Override
	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
		MqTransactionManager.context = applicationContext;

	}

	@Override
	public int getOrder() {
		return -100;
	}

}
